package com.ruoyi.network.edge;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Collection;
import java.util.Date;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import com.ruoyi.network.common.Constant;
import com.ruoyi.network.common.NPSocket;
import com.ruoyi.network.config.AbstractHelper;
import com.ruoyi.network.util.Threads;

import lombok.extern.slf4j.Slf4j;


@Slf4j
public class EdgeHelper extends AbstractHelper{
	// 监控线程
	public static ScheduledExecutorService monitor = Executors.newScheduledThreadPool(1);
	
	public static void start() {
		monitor.scheduleWithFixedDelay(() -> {
			clear();
		}, 15, 15, TimeUnit.SECONDS);

	}
	public synchronized static void distory() {
		Threads.shutdownAndAwaitTermination(monitor);
		closeClients(EdgeContext.all.keySet());
	}
	/**
	 * @param clients 
	 * 
	 */
	public synchronized static void closeClients(Collection<NPSocket> clients) {
		for (NPSocket clientSocket : clients) {
			try {
				close(clientSocket);
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
	}
	//synchronized是为了保证 dispatch 与 clear必须是互斥操作
	private synchronized static void clear() {
		List<NPSocket> timeouts = new LinkedList<>();
		log.debug("开始扫描失效与超时连接");
		Date start = new Date();
		for (NPSocket clientSocket : EdgeContext.all.keySet()) {
			Date lastTime = EdgeContext.all.get(clientSocket);
			Date now = new Date();
			Date createTime = clientSocket.getCreateTime();
			boolean isOpen = clientSocket.getSocketChannel().isOpen();
			boolean isReadTimeout = comparseSec(lastTime, now) >= Constant.READ_TIMEOUT;
			boolean isBindTimeout = !clientSocket.isBind()
					&& comparseSec(createTime, now) >= Constant.BIND_TIMEOUT;
			if (!isOpen || isReadTimeout || isBindTimeout) {
				timeouts.add(clientSocket);
			}

		}
//		log.debug(
//				"超时连接找到" + timeouts.size() + "个,用时 : " + (new Date().getTime() - start.getTime()) + "毫秒");
//		log.debug("开始清除失效与超时连接");
		closeClients(timeouts);
//		show();

	}
	
	
	public static void show() {
		log.debug("pairs " + EdgeContext.pairs.size());
		log.debug("all " + EdgeContext.all.size());
		log.debug("inners " + EdgeContext.inners.size());
		log.debug("outers " + EdgeContext.outers.size());
		log.debug("clients " + EdgeContext.clients.size());

	}

	public static void handleConnect(String name, Selector selector, SelectionKey key, Date timestamp)
			throws Exception {
		SocketChannel socketChannel = (SocketChannel) key.channel();
		NPSocket socket = new NPSocket(name,selector, socketChannel, false);
		EdgeContext.all.put(socket, timestamp);
		if (name.equals("外网连接")) {
			EdgeContext.outers.put(socket, timestamp);
		} else {
			EdgeContext.inners.put(socket, timestamp);
		}
		// 不管连接情况,统一对所有未绑定的对象同步非阻塞分配,
		Iterator<NPSocket> it = EdgeContext.outers.keySet().iterator();
		while (it.hasNext()) {
			NPSocket tmp = it.next();
			dispatch(selector, tmp);
		}

	}
	
	//synchronized是为了保证 dispatch 与 clear必须是互斥操作
	private synchronized static void dispatch(Selector selector, NPSocket outerSocket) {
		if (!outerSocket.isBind() && outerSocket.getSocketChannel().isOpen()) {
			Iterator<NPSocket> it = EdgeContext.inners.keySet().iterator();
			while (it.hasNext()) {
				NPSocket innerSocket = it.next();
				if (!innerSocket.isBind() && outerSocket.getSocketChannel().isOpen() && !outerSocket.isBind()) {
					EdgeContext.pairs.put(outerSocket.getSocketChannel(), innerSocket);
					EdgeContext.pairs.put(innerSocket.getSocketChannel(), outerSocket);
					outerSocket.setBind(true);
					innerSocket.setBind(true);
					try {
						// 绑定完毕后才能注册为读,以保证报文顺序正确
						// register()方法是阻塞的,必须要selector唤醒才能返回
						// 绑定完毕后才能注册为读,以保证报文顺序正确
						long start = System.currentTimeMillis();
						outerSocket.getSocketChannel().register(selector, SelectionKey.OP_READ);
						innerSocket.getSocketChannel().register(selector, SelectionKey.OP_READ);
						long now = System.currentTimeMillis();
						log.debug("bind success ... " + (now - start));
						break;
					} catch (Exception e) {
						e.printStackTrace();
						try {
							close(outerSocket);
						} catch (Exception e1) {
							e1.printStackTrace();
						}
					}
				}
			}
		}
	}

	private static void close(NPSocket socket1) throws Exception {
		SocketChannel channel1 = socket1.getSocketChannel();
		NPSocket socket2 = EdgeContext.pairs.get(channel1);
		if (socket2 != null) {
			SocketChannel channel2 = socket2.getSocketChannel();
			EdgeContext.pairs.remove(channel1);
			EdgeContext.all.remove(socket2);
			EdgeContext.inners.remove(socket2);
			EdgeContext.outers.remove(socket2);
			channel2.close();
			EdgeContext.pairs.remove(channel2);
		}
		if (socket1 != null) {
			EdgeContext.all.remove(socket1);
			EdgeContext.inners.remove(socket1);
			EdgeContext.outers.remove(socket1);
			// 必须要将close放在最后处理,不然Map中找不到对应的value
			channel1.close();
		}
	}

	public static void close(Selector selector, SelectionKey key) {
		try {
			if (key != null) {
				key.cancel();
				SocketChannel channel1 = (SocketChannel) key.channel();
				if (channel1 != null) {
					close(channel1);
				}
			}
			if (selector != null) {
				try {
					selector.close();
				} catch (IOException e) {
					e.printStackTrace();
				}
			}

		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	private static void close(SocketChannel channel1) throws IOException {
		NPSocket socket2 = EdgeContext.pairs.get(channel1);
		if (socket2 != null) {
			SocketChannel channel2 = socket2.getSocketChannel();
			NPSocket socket1 = EdgeContext.pairs.get(channel2);

			EdgeContext.pairs.remove(channel2);
			EdgeContext.all.remove(socket2);
			EdgeContext.inners.remove(socket2);
			EdgeContext.outers.remove(socket2);
			if (socket1 != null) {
				EdgeContext.pairs.remove(channel1);
				EdgeContext.all.remove(socket1);
				EdgeContext.inners.remove(socket1);
				EdgeContext.outers.remove(socket1);
			}
			// 必须要将close放在最后处理,不然Map中找不到对应的value
			channel2.close();
		}
		channel1.close();
	}

	public static void handleRead(String name, Selector selector, SelectionKey key, Date timestamp) throws Exception {
		SocketChannel channel1 = (SocketChannel) key.channel();
		NPSocket socket1 = EdgeContext.pairs.get(channel1);
		if (socket1 != null) {// 绑定了转发Socket才能读数据
			// 创建ByteBuffer，并开辟一个1M的缓冲区
			ByteBuffer buffer1 = ByteBuffer.allocate(1 * 1024);
			// 读取请求码流，返回读取到的字节数
			int readBytes = channel1.read(buffer1);
			// 读取到字节，对字节进行编解码
			if (readBytes > 0) {
				NPSocket socket2 = EdgeContext.pairs.get(socket1.getSocketChannel());
				// 方式一:直接发送
				buffer1.flip();
				
				//使用nio传输文件需要注意的是会出现粘包和服务器端缓冲区满的情况
				//第一种情况，客户端发送30次数据，而服务器端只接收到18次的情况，这种情况出现主要是服务器端是以流的方式接收数据，它并不知道每次客户端传输数据的大小而造成的
				//第二种情况是服务器端缓冲区满，导致客户端数据传输失败，这种情况下，需要判断传输int send = client.write(sendBuffer)的send值，如果send值为0,则服务器端的数据缓冲区可能满了
				//在上传大文件时异常,原因:在BIO中,当缓冲区满了就阻塞了,而对于非阻塞的NIO,缓冲区满了就立即返回0
				int isOk = socket1.getSocketChannel().write(buffer1);
				//服务器端可能因为缓存区满，而导致数据传输失败，需要重新发送
				while(isOk == 0){
					Thread.sleep(100L);
					isOk = socket1.getSocketChannel().write(buffer1);
				}
				buffer1.clear();

				// 更新通信时间
				EdgeContext.all.put(socket1, timestamp);
				EdgeContext.all.put(socket2, timestamp);
			}
			// 连接已经关闭，释放资源
			else if (readBytes < 0) {
				EdgeHelper.close(null, key);
			}
			
		}

	}

}
